{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction: Partitioning Data \n",
    "\n",
    "Problem: we have a large dataset that we want to partition into smaller sections in order to run a feature engineering pipeline over these partitions in parallel.\n",
    "Approach: divide the data into partitions by hashing the customer id to an integer and then modolu (integer) dividing by the number of partitions.\n",
    "\n",
    "After determining the partition for each customer id, we can iterate over the data and write the partitions to the correct directory. At the end of the process, we'll have `N_PARTITIONS` of customer data, each containing all the data for a subset of customers. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import hashlib\n",
    "\n",
    "from IPython.core.interactiveshell import InteractiveShell\n",
    "InteractiveShell.ast_node_interactivity = \"all\"\n",
    "\n",
    "N_PARTITIONS = 1000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Hashing\n",
    "\n",
    "Hashing is the process of mapping a string of characters to a fixed length value or key. For this problem, we'll take a string, the customer id (`msno` in the data) and map it to an integer (16 bits). To convert to a partition number, we then integer (modulo) divide this integer by the number of partitions.\n",
    "\n",
    "### Consistency of Hashing\n",
    "\n",
    "To make sure that the integer for a given string is always the same, we use the `md5` algorithm form the `hashlib` library. There is no way to go backwards from an integer to the string, but we can always go from the string to the same exact hash. Since we don't need to partition our customers in any particular groupings, randomly assigning customers to partitions is an acceptable approach."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def id_to_hash(customer_id):\n",
    "    \"\"\"Return a 16-bit integer hash of a customer id string\"\"\"\n",
    "    return int(hashlib.md5(customer_id.encode('utf-8')).hexdigest(), 16)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Rb9UwLQTrxzBVwCB6+bCcSQWZ9JiNLC9dXtM1oEsZA8=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>11</td>\n",
       "      <td>20110911</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  Rb9UwLQTrxzBVwCB6+bCcSQWZ9JiNLC9dXtM1oEsZA8=     1   0     NaN   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0              11                20110911  "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>payment_method_id</th>\n",
       "      <th>payment_plan_days</th>\n",
       "      <th>plan_list_price</th>\n",
       "      <th>actual_amount_paid</th>\n",
       "      <th>is_auto_renew</th>\n",
       "      <th>transaction_date</th>\n",
       "      <th>membership_expire_date</th>\n",
       "      <th>is_cancel</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>YyO+tlZtAXYXoZhNr3Vg3+dfVQvrBVGO8j1mfqe4ZHc=</td>\n",
       "      <td>41</td>\n",
       "      <td>30</td>\n",
       "      <td>129</td>\n",
       "      <td>129</td>\n",
       "      <td>1</td>\n",
       "      <td>20150930</td>\n",
       "      <td>20151101</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  payment_method_id  \\\n",
       "0  YyO+tlZtAXYXoZhNr3Vg3+dfVQvrBVGO8j1mfqe4ZHc=                 41   \n",
       "\n",
       "   payment_plan_days  plan_list_price  actual_amount_paid  is_auto_renew  \\\n",
       "0                 30              129                 129              1   \n",
       "\n",
       "   transaction_date  membership_expire_date  is_cancel  \n",
       "0          20150930                20151101          0  "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>date</th>\n",
       "      <th>num_25</th>\n",
       "      <th>num_50</th>\n",
       "      <th>num_75</th>\n",
       "      <th>num_985</th>\n",
       "      <th>num_100</th>\n",
       "      <th>num_unq</th>\n",
       "      <th>total_secs</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>rxIP2f2aN0rYNp+toI0Obt/N/FYQX8hcO1fTmmy2h34=</td>\n",
       "      <td>20150513</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>280.335</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno      date  num_25  num_50  \\\n",
       "0  rxIP2f2aN0rYNp+toI0Obt/N/FYQX8hcO1fTmmy2h34=  20150513       0       0   \n",
       "\n",
       "   num_75  num_985  num_100  num_unq  total_secs  \n",
       "0       0        0        1        1     280.335  "
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>is_churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>waLDQMmcOu2jLDaV1ddDkgCrB/jl6sD66Xzs0Vqax1Y=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  is_churn\n",
       "0  waLDQMmcOu2jLDaV1ddDkgCrB/jl6sD66Xzs0Vqax1Y=         1"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>is_churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>4n+fXlyJvfQnTeKXTWT507Ll4JVYGrOC8LHCfwBmPE4=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  is_churn\n",
       "0  4n+fXlyJvfQnTeKXTWT507Ll4JVYGrOC8LHCfwBmPE4=         0"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "members = pd.read_csv('/data/churn/members_v3.csv', nrows = 1)\n",
    "members\n",
    "\n",
    "transactions = pd.read_csv('/data/churn/all_trans.csv', nrows = 1)\n",
    "transactions\n",
    "\n",
    "logs = pd.read_csv('/data/churn/user_logs.csv', nrows = 1)\n",
    "logs\n",
    "\n",
    "train = pd.read_csv('/data/churn/all_train.csv', nrows = 1)\n",
    "train\n",
    "\n",
    "test = pd.read_csv('/data/churn/sample_submission_v2.csv', nrows = 1)\n",
    "test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "209512247756457468966515739358104959027"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/plain": [
       "27"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "id_to_hash(members.loc[0, 'msno'])\n",
    "id_to_hash(members.loc[0, 'msno']) % N_PARTITIONS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "311407269432611323870693642675616983728"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "text/plain": [
       "728"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "id_to_hash(transactions.loc[0, 'msno'])\n",
    "id_to_hash(transactions.loc[0, 'msno']) % N_PARTITIONS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following code creates `N_PARTITIONS` empty directories. The files within each directory will be named exactly the same so the directory name can be used to distinguish partitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import os\n",
    "\n",
    "base_dir = '/data/churn/partitions/'\n",
    "\n",
    "if not os.path.exists(base_dir + 'p999'):\n",
    "    # Create a new directory for each partition\n",
    "    for i in range(N_PARTITIONS):\n",
    "        os.makedirs(base_dir + f'p{i}', exist_ok=False)\n",
    "    \n",
    "len(os.listdir(base_dir))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Files\n",
    "\n",
    "In each partition there are 5 csv files. \n",
    "\n",
    "* `transactions.csv`\n",
    "* `train.csv`\n",
    "* `test.csv`\n",
    "* `members.csv`\n",
    "* `logs.csv`\n",
    "\n",
    "The following code writes the header for each of the five files in each of the `N_PARTITION` partitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'msno,payment_method_id,payment_plan_days,plan_list_price,actual_amount_paid,is_auto_renew,transaction_date,membership_expire_date,is_cancel'"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "InteractiveShell.ast_node_interactivity = 'last_expr'\n",
    "','.join(list(transactions.columns))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_blank_partitions():\n",
    "    \"\"\"Create blank files in each partition and write the file header\"\"\"\n",
    "    # For each partition create the files with headers\n",
    "    for i in range(N_PARTITIONS):\n",
    "        directory = base_dir + f'p{i}/'\n",
    "        # Create five files\n",
    "        for file in ['transactions.csv', 'train.csv', 'test.csv', 'members.csv', 'logs.csv']:\n",
    "            # Write file header as first line\n",
    "            with open(directory + file, 'w') as f:\n",
    "                if file == 'transactions.csv':\n",
    "                    f.write(','.join(list(transactions.columns)))\n",
    "                elif file == 'train.csv':\n",
    "                    f.write(','.join(list(train.columns)))\n",
    "                elif file == 'test.csv':\n",
    "                    f.write(','.join(list(train.columns)))\n",
    "                elif file == 'members.csv':\n",
    "                    f.write(','.join(list(members.columns)))\n",
    "                elif file == 'logs.csv':\n",
    "                    f.write(','.join(list(logs.columns)))\n",
    "                    \n",
    "    return directory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['logs.csv', 'members.csv', 'train.csv', 'test.csv', 'transactions.csv']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "directory = create_blank_partitions()\n",
    "os.listdir(directory)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When it comes time to write the data, we'll open the existing files using append (`a`) option and add to whatever is there. At this point, each file just has a header."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: [msno, city, bd, gender, registered_via, registration_init_time]\n",
       "Index: []"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(directory + 'members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example of Writing a Row\n",
    "\n",
    "For each file, one option to write the data to the partitions is to iterate over the rows one at a time. The process for handling a row is:\n",
    "\n",
    "1. Convert the customer id to an integer by hashing\n",
    "2. Convert the integer to a partition number by modulus dividing by the number of partitions\n",
    "3. Append the row to the correct partition directory and file\n",
    "\n",
    "Let's see how this works with a single row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Iterate through the dataframe one row at a time\n",
    "for i, row in members.iterrows():\n",
    "    # Find the partition by hashing the id\n",
    "    partition = id_to_hash(row['msno']) % N_PARTITIONS\n",
    "    # Open the file for appending\n",
    "    with open(base_dir + f'p{partition}/members.csv', 'a') as f:\n",
    "        # Write a newline and then the information\n",
    "        f.write('\\n')\n",
    "        f.write(','.join([str(x) for x in row.values]))\n",
    "    if i > 1:\n",
    "        break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Rb9UwLQTrxzBVwCB6+bCcSQWZ9JiNLC9dXtM1oEsZA8=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>11</td>\n",
       "      <td>20110911</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  Rb9UwLQTrxzBVwCB6+bCcSQWZ9JiNLC9dXtM1oEsZA8=     1   0     NaN   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0              11                20110911  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Everything looks like it went well with the first attempt. However, we might want to ask if iterating over the dataset one row at a time using `iterrows` is the quickest options."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Performance of Different Methods\n",
    "\n",
    "There are a number of different options to handle writing the data to the correct partition. To find out which method is the best, we'll try 4 approaches:\n",
    "\n",
    "1. `df.iterrows()`: iterate through the dataframe one row at a time with rows represented as series\n",
    "2. `df.itertuples()`: iterate through the dataframe one row at a time with rows represented as tuples\n",
    "3. `df.apply()`: iterate through the data one row at a time using `apply`\n",
    "4. `groupby(partition)` and save each group with `to_csv()`: iterate through the dataframe one partition at a time\n",
    "\n",
    "The four approaches have different applicability and performance characteristics (see this [Stack Overflow](https://stackoverflow.com/questions/24870953/does-iterrows-have-performance-issues/24871316#24871316) answer). The way to find out which one is the quickest is to try them all. This isn't meant to represent all use cases, so your particular results may vary."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from timeit import default_timer as timer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Member Information\n",
    "\n",
    "We'll start off with the members. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(6769473, 6)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "members = pd.read_csv('/data/churn/members_v3.csv')\n",
    "members.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Iterrows\n",
    "\n",
    "The first implementation to try is `iterrows`. This is fairly slow because Pandas packages the row as a Pandas series before iteration. However, it does allow us to access each value using conventional locating. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing 6769472 lines took 557 seconds using iterrows.\n"
     ]
    }
   ],
   "source": [
    "start = timer()\n",
    "\n",
    "for i, row in members.iterrows():\n",
    "    # Find the partition number by hashing the id\n",
    "    partition = id_to_hash(row['msno']) % N_PARTITIONS\n",
    "    \n",
    "    # Open file for appending\n",
    "    with open(base_dir + f'p{partition}/members.csv', 'a') as f:\n",
    "        # Write a new line and then data\n",
    "        f.write('\\n')\n",
    "        f.write(','.join([str(x) for x in row.values]))\n",
    "        \n",
    "    if i % 10000 == 0:\n",
    "        print(f'{100 * round(i / members.shape[0], 2)}% complete. {round(timer() - start)} seconds elapsed.', end = '\\r')\n",
    "\n",
    "end = timer()\n",
    "print(f'Processing {i} lines took {round(end - start)} seconds using iterrows.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>+zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>9</td>\n",
       "      <td>20161227</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=</td>\n",
       "      <td>5</td>\n",
       "      <td>26</td>\n",
       "      <td>male</td>\n",
       "      <td>3</td>\n",
       "      <td>20141109</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>7</td>\n",
       "      <td>20161228</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=</td>\n",
       "      <td>1</td>\n",
       "      <td>23</td>\n",
       "      <td>female</td>\n",
       "      <td>4</td>\n",
       "      <td>20170113</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4</td>\n",
       "      <td>20170121</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  +zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=     1   0     NaN   \n",
       "1  3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=     5  26    male   \n",
       "2  BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=     1   0     NaN   \n",
       "3  qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=     1  23  female   \n",
       "4  c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=     1   0     NaN   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0               9                20161227  \n",
       "1               3                20141109  \n",
       "2               7                20161228  \n",
       "3               4                20170113  \n",
       "4               4                20170121  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This approach works but is fairly slow."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Itertuples\n",
    "\n",
    "Itertuples should be faster than iterrows because Pandas packages the row as a tuple instead of as a series. The tradeoff is we need to be careful when accessing the elements of the series since we can't refer to them by name. For example, to make sure we are hashing the customer id (`msno`) we need to grab the second element of the tuple. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = create_blank_partitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing 6769472 lines took 138 seconds using itertuples.\n"
     ]
    }
   ],
   "source": [
    "start = timer()\n",
    "\n",
    "for i, tup in enumerate(members.itertuples()):\n",
    "    \n",
    "    # Find the partition number by hashing the id\n",
    "    partition = id_to_hash(tup[1]) % N_PARTITIONS\n",
    "    \n",
    "    # Open file for appending\n",
    "    with open(base_dir + f'p{partition}/members.csv', 'a') as f:\n",
    "        # Write a new line and then data\n",
    "        f.write('\\n')\n",
    "        f.write(','.join([str(x) for x in tup[1:]]))\n",
    "        \n",
    "    if i % 10000 == 0:\n",
    "        print(f'{100 * round(i / members.shape[0], 2)}% complete. {round(timer() - start)} seconds elapsed.', end = '\\r')\n",
    "\n",
    "end = timer()\n",
    "print(f'Processing {i} lines took {round(end - start)} seconds using itertuples.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>+zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>9</td>\n",
       "      <td>20161227</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=</td>\n",
       "      <td>5</td>\n",
       "      <td>26</td>\n",
       "      <td>male</td>\n",
       "      <td>3</td>\n",
       "      <td>20141109</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>7</td>\n",
       "      <td>20161228</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=</td>\n",
       "      <td>1</td>\n",
       "      <td>23</td>\n",
       "      <td>female</td>\n",
       "      <td>4</td>\n",
       "      <td>20170113</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4</td>\n",
       "      <td>20170121</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  +zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=     1   0     NaN   \n",
       "1  3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=     5  26    male   \n",
       "2  BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=     1   0     NaN   \n",
       "3  qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=     1  23  female   \n",
       "4  c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=     1   0     NaN   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0               9                20161227  \n",
       "1               3                20141109  \n",
       "2               7                20161228  \n",
       "3               4                20170113  \n",
       "4               4                20170121  "
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This approach was much faster because Pandas does not have to convert each row into a Series which has more overhead than a tuple."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = create_blank_partitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Apply\n",
    "\n",
    "Another option would be to use the apply over the rows. To use `apply`, we write a small function that saves the row and then call apply to the dataframe using `axis = 1` which sends each row to the function. This also sends a row as a Series to the function, but in practice it seems to be much faster than `iterrows`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_row(row, name):\n",
    "    # Find the partition number by hashing the id\n",
    "    partition = id_to_hash(row['msno']) % N_PARTITIONS\n",
    "    \n",
    "    # Open file for appending\n",
    "    with open(base_dir + f'p{partition}/{name}.csv', 'a') as f:\n",
    "        # Write a new line and then data\n",
    "        f.write('\\n')\n",
    "        f.write(','.join([str(x) for x in row.values]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 6769473/6769473 [04:41<00:00, 24048.87it/s]"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing 6769473 rows took 282 seconds using apply.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "from tqdm import tqdm_notebook\n",
    "from tqdm import tqdm\n",
    "tqdm.pandas()\n",
    "\n",
    "start = timer()\n",
    "members.progress_apply(save_row, axis = 1, name = 'members')\n",
    "end = timer()\n",
    "\n",
    "print(f'Processing {members.shape[0]} rows took {round(end - start)} seconds using apply.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>+zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>9</td>\n",
       "      <td>20161227</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=</td>\n",
       "      <td>5</td>\n",
       "      <td>26</td>\n",
       "      <td>male</td>\n",
       "      <td>3</td>\n",
       "      <td>20141109</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>7</td>\n",
       "      <td>20161228</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=</td>\n",
       "      <td>1</td>\n",
       "      <td>23</td>\n",
       "      <td>female</td>\n",
       "      <td>4</td>\n",
       "      <td>20170113</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4</td>\n",
       "      <td>20170121</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  +zMKqjvsTvD7O0Fvntk3VXe4ovwvD4KYk6PJZ92Ky60=     1   0     NaN   \n",
       "1  3p2AY1tZAYa4LFcs0/plkuPv2hY9smh/xgcbKjtU9Dc=     5  26    male   \n",
       "2  BO0XUBzHeItkHI3N5g4uL08Ld1T/ZW/8GrbjBmT4s3w=     1   0     NaN   \n",
       "3  qy7PNK2EE4+x6xeIdqjFVw5FlmxnFKylv6LKqqGbSo4=     1  23  female   \n",
       "4  c36721uHBQyhoVko21J9rR44Fex2ul72a74k0M7IkiQ=     1   0     NaN   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0               9                20161227  \n",
       "1               3                20141109  \n",
       "2               7                20161228  \n",
       "3               4                20170113  \n",
       "4               4                20170121  "
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So `apply` is faster than `iterrows` but slower than `itertuples` (at least in this case). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = create_blank_partitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Groupby \n",
    "\n",
    "The final option we'll try is to group the data by the partition after converting all the customer ids into partition numbers in once operation.\n",
    "\n",
    "1. Compute the partitions with the hashing function all at once\n",
    "2. Groupby the partition\n",
    "3. Write the grouped dataframe to the correct partition directory and file\n",
    "\n",
    "To find out the quickest way to convert all the customer ids to an integer, we can compare `map` and `apply`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8.02 s ± 82.3 ms per loop (mean ± std. dev. of 3 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit -n 1 -r 3\n",
    "members['msno'].map(id_to_hash) % 1000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "7.87 s ± 9.26 ms per loop (mean ± std. dev. of 3 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit -n 1 -r 3\n",
    "members['msno'].apply(id_to_hash) % 1000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It looks like `apply` is slightly faster although the difference is not significant. We'll convert all the customer ids to the partitions once so this is not a large time cost. Before we go on, we should make sure that the hashing and conversion to partition operation is creating partitions close to the same size."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "689    7027\n",
       "842    7012\n",
       "844    6992\n",
       "660    6986\n",
       "91     6983\n",
       "Name: partition, dtype: int64"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "members['partition'] = members['msno'].apply(id_to_hash) % 1000\n",
    "members['partition'].value_counts().head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "count    1000.000000\n",
       "mean     6769.473000\n",
       "std        81.213231\n",
       "min      6494.000000\n",
       "25%      6716.000000\n",
       "50%      6773.000000\n",
       "75%      6825.000000\n",
       "max      7027.000000\n",
       "Name: partition, dtype: float64"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "members['partition'].value_counts().describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It looks like the number of members in each partition is fairly constant. We can check another dataset to make sure."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "count     1000.00000\n",
       "mean     22978.75500\n",
       "std        652.76208\n",
       "min      20893.00000\n",
       "25%      22532.75000\n",
       "50%      22975.00000\n",
       "75%      23420.25000\n",
       "max      25353.00000\n",
       "Name: partition, dtype: float64"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "transactions = pd.read_csv('/data/churn/all_trans.csv')\n",
    "transactions['partition'] = transactions['msno'].apply(id_to_hash) % 1000\n",
    "transactions['partition'].value_counts().describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following cell runs the group by approach to partitioning the data. The biggest thing to watch out for is making sure we are appending to the file each time (open with `a`). When we write to a csv with `to_csv`, we can pass in an already open file. We also don't write a header since we already created the headers in every file, and do not write the index."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Processing 6769473 rows took 41 seconds using groupby.\n"
     ]
    }
   ],
   "source": [
    "start = timer()\n",
    "members['partition'] = members['msno'].apply(id_to_hash) % N_PARTITIONS\n",
    "\n",
    "# Iteration through grouped partitions\n",
    "for partition, grouped in members.groupby('partition'):\n",
    "    \n",
    "    grouped = grouped.drop(columns = 'partition')\n",
    "    # Open file for appending\n",
    "    with open(base_dir + f'p{partition}/members.csv', 'a') as f:\n",
    "        f.write('\\n')\n",
    "        grouped.to_csv(f, header = False, index = False)\n",
    "        \n",
    "end = timer()\n",
    "print(f'Processing {members.shape[0]} rows took {round(end - start)} seconds using groupby.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>bpIibSSY6wymQbGaQOR9q6dcWKg7lUfw3Y+LttzAQNQ=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4</td>\n",
       "      <td>20170104</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>HZwqy9brMyBDuFVpXlAqli8yoAixLc1rA0ExAZYZR50=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>male</td>\n",
       "      <td>3</td>\n",
       "      <td>20130220</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>kW0/xDZUihRKFMa3ti+vq3fF/O2li5aYpY+szvzg0ko=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>3</td>\n",
       "      <td>20130227</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>yCUq5TNkbcJF0inE45ICYI//gZ+FzPmwSZWmFie4nk8=</td>\n",
       "      <td>6</td>\n",
       "      <td>31</td>\n",
       "      <td>female</td>\n",
       "      <td>9</td>\n",
       "      <td>20130305</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>VsM62mNuBRPH2YZSZKaRlD0IQsqoJa55aKxukV84oY4=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>female</td>\n",
       "      <td>3</td>\n",
       "      <td>20150213</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  bpIibSSY6wymQbGaQOR9q6dcWKg7lUfw3Y+LttzAQNQ=     1   0     NaN   \n",
       "1  HZwqy9brMyBDuFVpXlAqli8yoAixLc1rA0ExAZYZR50=     1   0    male   \n",
       "2  kW0/xDZUihRKFMa3ti+vq3fF/O2li5aYpY+szvzg0ko=     1   0     NaN   \n",
       "3  yCUq5TNkbcJF0inE45ICYI//gZ+FzPmwSZWmFie4nk8=     6  31  female   \n",
       "4  VsM62mNuBRPH2YZSZKaRlD0IQsqoJa55aKxukV84oY4=     1   0  female   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0               4                20170104  \n",
       "1               3                20130220  \n",
       "2               3                20130227  \n",
       "3               9                20130305  \n",
       "4               3                20150213  "
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The group by approach to partitioning the data is by far the fastest method. We'll put this into a function to use with all the datasets."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reusable Hashing Dataframe Function\n",
    "\n",
    "To make the process reusable, we'll write a function that does this for us. It will take in a dataframe, a name for the file to save the data to, and an optional progress argument. The function will map the customer id (`msno`) to a partition number using the hash modulo the number of partitions, group the dataframe by the partition, and write the grouped dataframe to the appropriate directory. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "members = pd.read_csv('/data/churn/members_v3.csv', nrows = 1)\n",
    "transactions = pd.read_csv('/data/churn/all_trans.csv', nrows = 1)\n",
    "logs = pd.read_csv('/data/churn/user_logs.csv', nrows = 1)\n",
    "train = pd.read_csv('/data/churn/all_train.csv', nrows = 1)\n",
    "test = pd.read_csv('/data/churn/sample_submission_v2.csv', nrows = 1)\n",
    "\n",
    "_ = create_blank_partitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: [msno, city, bd, gender, registered_via, registration_init_time]\n",
       "Index: []"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv('/data/churn/partitions/p999/members.csv').head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "def partition_by_hashing(df, name, progress = None):\n",
    "    \"\"\"Partition a dataframe into N_PARTITIONS by hashing the id.\n",
    "    \n",
    "    Params\n",
    "    --------\n",
    "        df (pandas dataframe): dataframe for partition. Must have 'msno' column.\n",
    "        name (str): name of dataframe. Used for saving the row data.\n",
    "        progress (int, optional): number of rows to be processed before displaying information.\n",
    "                                  Defaults to None\n",
    "                                  \n",
    "    Returns:\n",
    "    --------\n",
    "        Nothing returned. Dataframe is saved one line at a time as csv files to the N_PARTITIONS \n",
    "    \"\"\"\n",
    "    start = timer()\n",
    "    \n",
    "    # Map the customer id to a partition number\n",
    "    df['partition'] = df['msno'].apply(id_to_hash) % N_PARTITIONS\n",
    "    \n",
    "    # Iterate through one row at a time\n",
    "    for partition, grouped in df.groupby('partition'):\n",
    "        \n",
    "        # Don't need to save the partition column\n",
    "        grouped = grouped.drop(columns = 'partition')\n",
    "        \n",
    "        # Open file for appending\n",
    "        with open(base_dir + f'p{partition}/{name}.csv', 'a') as f:\n",
    "            # Write a new line and then data\n",
    "            f.write('\\n')\n",
    "            grouped.to_csv(f, header = False, index = False)\n",
    "            \n",
    "        # Record progress every `progress` steps\n",
    "        if progress is not None:\n",
    "            if partition % progress == 0:\n",
    "                print(f'{100 * round(partition / N_PARTITIONS, 2)}% complete. {round(timer() - start)} seconds elapsed.', end = '\\r')\n",
    "    \n",
    "    end = timer()\n",
    "    if progress is not None:\n",
    "        print(f'\\n{df.shape[0]} rows processed in {round(end - start)} seconds.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "99.0% complete. 41 seconds elapsed.onds elapsed..\n",
      "6769473 rows processed in 41 seconds.\n"
     ]
    }
   ],
   "source": [
    "members = pd.read_csv('/data/churn/members_v3.csv')\n",
    "partition_by_hashing(members, name = 'members', progress = 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>gender</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>registration_init_time</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>bpIibSSY6wymQbGaQOR9q6dcWKg7lUfw3Y+LttzAQNQ=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>4</td>\n",
       "      <td>20170104</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>HZwqy9brMyBDuFVpXlAqli8yoAixLc1rA0ExAZYZR50=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>male</td>\n",
       "      <td>3</td>\n",
       "      <td>20130220</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>kW0/xDZUihRKFMa3ti+vq3fF/O2li5aYpY+szvzg0ko=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>3</td>\n",
       "      <td>20130227</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>yCUq5TNkbcJF0inE45ICYI//gZ+FzPmwSZWmFie4nk8=</td>\n",
       "      <td>6</td>\n",
       "      <td>31</td>\n",
       "      <td>female</td>\n",
       "      <td>9</td>\n",
       "      <td>20130305</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>VsM62mNuBRPH2YZSZKaRlD0IQsqoJa55aKxukV84oY4=</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>female</td>\n",
       "      <td>3</td>\n",
       "      <td>20150213</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  city  bd  gender  \\\n",
       "0  bpIibSSY6wymQbGaQOR9q6dcWKg7lUfw3Y+LttzAQNQ=     1   0     NaN   \n",
       "1  HZwqy9brMyBDuFVpXlAqli8yoAixLc1rA0ExAZYZR50=     1   0    male   \n",
       "2  kW0/xDZUihRKFMa3ti+vq3fF/O2li5aYpY+szvzg0ko=     1   0     NaN   \n",
       "3  yCUq5TNkbcJF0inE45ICYI//gZ+FzPmwSZWmFie4nk8=     6  31  female   \n",
       "4  VsM62mNuBRPH2YZSZKaRlD0IQsqoJa55aKxukV84oY4=     1   0  female   \n",
       "\n",
       "   registered_via  registration_init_time  \n",
       "0               4                20170104  \n",
       "1               3                20130220  \n",
       "2               3                20130227  \n",
       "3               9                20130305  \n",
       "4               3                20150213  "
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/members.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training Data\n",
    "\n",
    "Now we can use this function to partition the training data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "99.0% complete. 8 seconds elapsed.onds elapsed..\n",
      "1963891 rows processed in 8 seconds.\n"
     ]
    }
   ],
   "source": [
    "train = pd.read_csv('/data/churn/all_train.csv')\n",
    "partition_by_hashing(train, name = 'train', progress = 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>is_churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Ad9xf2W6ID3zguduv1lKdla80V/iT2cFWbDxIcEMQOs=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>nL5mKfbpD9mjLJiIbOa1MsrHXKlmC4Nt5S3ieKtqUq8=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>TmSYOz2xCdk5j1hmUP72/FQdkU3kpoqEl9RDc9UBtf8=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>NNYVPCEq8Pk2QYIGSBSIiO+XcX/Sqa2TG5+szr4DMuk=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>bhzedCaoyawwHOssUE6IXI1BP4I0/4nDC1H6CtXcIK4=</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  is_churn\n",
       "0  Ad9xf2W6ID3zguduv1lKdla80V/iT2cFWbDxIcEMQOs=         1\n",
       "1  nL5mKfbpD9mjLJiIbOa1MsrHXKlmC4Nt5S3ieKtqUq8=         1\n",
       "2  TmSYOz2xCdk5j1hmUP72/FQdkU3kpoqEl9RDc9UBtf8=         1\n",
       "3  NNYVPCEq8Pk2QYIGSBSIiO+XcX/Sqa2TG5+szr4DMuk=         1\n",
       "4  bhzedCaoyawwHOssUE6IXI1BP4I0/4nDC1H6CtXcIK4=         1"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/train.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Testing Data\n",
    "\n",
    "The nice thing about a function is we can keep applying it, changing only the arguments! "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "99.0% complete. 4 seconds elapsed.onds elapsed..\n",
      "907471 rows processed in 4 seconds.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>is_churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>bFTbnI7GC8TZJ2m9dfLF0bkFIEsy/0ERZzmFFekPWpY=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>SvdAOG3xp9glZIopNrlnOOiOLHdmnpHfwRGnNUpOrmo=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>GG+L3Jv6naoL8JjQYPIem7ISmtoHcMX453sqTSFcl2Q=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>OdT/fU9BKs+KrLeypHtoeQut97PBNWlDvjnMCUsJVzY=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>f0SvlQcRZBgZbOaoPBkj57xo81+GnlluK1rLyy0PQnc=</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  is_churn\n",
       "0  bFTbnI7GC8TZJ2m9dfLF0bkFIEsy/0ERZzmFFekPWpY=         0\n",
       "1  SvdAOG3xp9glZIopNrlnOOiOLHdmnpHfwRGnNUpOrmo=         0\n",
       "2  GG+L3Jv6naoL8JjQYPIem7ISmtoHcMX453sqTSFcl2Q=         0\n",
       "3  OdT/fU9BKs+KrLeypHtoeQut97PBNWlDvjnMCUsJVzY=         0\n",
       "4  f0SvlQcRZBgZbOaoPBkj57xo81+GnlluK1rLyy0PQnc=         0"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test = pd.read_csv('/data/churn/sample_submission_v2.csv')\n",
    "partition_by_hashing(test, name = 'test', progress = 10)\n",
    "pd.read_csv(base_dir + f'p{partition}/test.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transactional Data\n",
    "\n",
    "The second to last dataset is the customer transactions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "99.0% complete. 199 seconds elapsed.onds elapsed.\n",
      "22978755 rows processed in 201 seconds.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>payment_method_id</th>\n",
       "      <th>payment_plan_days</th>\n",
       "      <th>plan_list_price</th>\n",
       "      <th>actual_amount_paid</th>\n",
       "      <th>is_auto_renew</th>\n",
       "      <th>transaction_date</th>\n",
       "      <th>membership_expire_date</th>\n",
       "      <th>is_cancel</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>igTF6Ef1Y1chfDlEjV+59Hgp4mfh8ZNVXF/vlv1TvhY=</td>\n",
       "      <td>41</td>\n",
       "      <td>30</td>\n",
       "      <td>129</td>\n",
       "      <td>129</td>\n",
       "      <td>1</td>\n",
       "      <td>20160418</td>\n",
       "      <td>20160518</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>wlwPlei0VKJ6AF07YrL04gpbebUPitqtnqa3CnL4tmI=</td>\n",
       "      <td>41</td>\n",
       "      <td>30</td>\n",
       "      <td>149</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>20150107</td>\n",
       "      <td>20160106</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>m40NTt7XHbAVwCI4y6TYH0XBJdPeVVpBUSKT9NXP7xE=</td>\n",
       "      <td>39</td>\n",
       "      <td>30</td>\n",
       "      <td>149</td>\n",
       "      <td>149</td>\n",
       "      <td>1</td>\n",
       "      <td>20161231</td>\n",
       "      <td>20170210</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>wQ8y0ZtCUniH1rWDKyltgP+nfxDV9hAvjcFBa7l9uAU=</td>\n",
       "      <td>41</td>\n",
       "      <td>30</td>\n",
       "      <td>99</td>\n",
       "      <td>99</td>\n",
       "      <td>1</td>\n",
       "      <td>20160731</td>\n",
       "      <td>20160831</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>CZ87lCEDKmbk7EXCy6Kh0zopNJxoEBTfWq9Eq9Az+28=</td>\n",
       "      <td>32</td>\n",
       "      <td>195</td>\n",
       "      <td>894</td>\n",
       "      <td>894</td>\n",
       "      <td>0</td>\n",
       "      <td>20160726</td>\n",
       "      <td>20170209</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno  payment_method_id  \\\n",
       "0  igTF6Ef1Y1chfDlEjV+59Hgp4mfh8ZNVXF/vlv1TvhY=                 41   \n",
       "1  wlwPlei0VKJ6AF07YrL04gpbebUPitqtnqa3CnL4tmI=                 41   \n",
       "2  m40NTt7XHbAVwCI4y6TYH0XBJdPeVVpBUSKT9NXP7xE=                 39   \n",
       "3  wQ8y0ZtCUniH1rWDKyltgP+nfxDV9hAvjcFBa7l9uAU=                 41   \n",
       "4  CZ87lCEDKmbk7EXCy6Kh0zopNJxoEBTfWq9Eq9Az+28=                 32   \n",
       "\n",
       "   payment_plan_days  plan_list_price  actual_amount_paid  is_auto_renew  \\\n",
       "0                 30              129                 129              1   \n",
       "1                 30              149                   0              1   \n",
       "2                 30              149                 149              1   \n",
       "3                 30               99                  99              1   \n",
       "4                195              894                 894              0   \n",
       "\n",
       "   transaction_date  membership_expire_date  is_cancel  \n",
       "0          20160418                20160518          0  \n",
       "1          20150107                20160106          0  \n",
       "2          20161231                20170210          0  \n",
       "3          20160731                20160831          0  \n",
       "4          20160726                20170209          0  "
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "transactions = pd.read_csv('/data/churn/all_trans.csv')\n",
    "partition_by_hashing(transactions, name = 'transactions', progress = 10)\n",
    "pd.read_csv(base_dir + f'p{partition}/transactions.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## User Log Data\n",
    "\n",
    "The final dataset cannot be pass directly into the function because of the size which won't even let us read the entire file into memory. Instead, we can read it in a chunk at a time using Pandas and apply the function to each chunk. There are actually two log files of significantly different sizes, but we'll use the same chunking approach for each."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "30.514081415\n",
      "1.431465728\n"
     ]
    }
   ],
   "source": [
    "print(os.stat('/data/churn/user_logs.csv').st_size / 1e9)\n",
    "print(os.stat('/data/churn/user_logs_v2.csv').st_size / 1e9)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The second user logs can be processed in the previous manner because it can be read completely into memory, but we'll go ahead and apply the chunking method. The chunksize refers to the number of rows read in at a time. Using Pandas `read_csv` and specifying the `chunksize`, we can then iterate over the file one chunk at a time. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Overall time: 200 seconds.\n"
     ]
    }
   ],
   "source": [
    "chunksize = 1e6\n",
    "start = timer()\n",
    "\n",
    "for chunk in pd.read_csv('/data/churn/user_logs_v2.csv', chunksize = chunksize):\n",
    "    partition_by_hashing(chunk, name = 'logs', progress = None)\n",
    "    \n",
    "    if (i + 1) % 10 == 0:\n",
    "        print(f'{i * chunksize} rows processed.', end = '\\r')\n",
    "\n",
    "end = timer()\n",
    "print(f'\\nOverall time: {round(end - start)} seconds.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>date</th>\n",
       "      <th>num_25</th>\n",
       "      <th>num_50</th>\n",
       "      <th>num_75</th>\n",
       "      <th>num_985</th>\n",
       "      <th>num_100</th>\n",
       "      <th>num_unq</th>\n",
       "      <th>total_secs</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>555bSTll4Rzaz1vBg/VfGfaXKEE8S74EAdKPOCkBqpg=</td>\n",
       "      <td>20170301</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>25</td>\n",
       "      <td>26</td>\n",
       "      <td>6306.242</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>KORJwjTctoWENM9oM2Rrl432wxllmC1RvP5p84PUFI8=</td>\n",
       "      <td>20170316</td>\n",
       "      <td>6</td>\n",
       "      <td>1</td>\n",
       "      <td>2</td>\n",
       "      <td>3</td>\n",
       "      <td>36</td>\n",
       "      <td>34</td>\n",
       "      <td>10718.255</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>+GdZIQJdsQSeRKyu/GONhgWzK4R8Ufm59RpzwAj4OCE=</td>\n",
       "      <td>20170310</td>\n",
       "      <td>5</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>21</td>\n",
       "      <td>21</td>\n",
       "      <td>4948.548</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>9yZc5dve4sg96RQT1FyTPSmcDWrIVu+qsI7W7oOmFvE=</td>\n",
       "      <td>20170306</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>28</td>\n",
       "      <td>1</td>\n",
       "      <td>3436.411</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Rq9NMOw9RAbN3qUIG1MUU+vL0lqhAJUAsZiW0dsm/Lg=</td>\n",
       "      <td>20170322</td>\n",
       "      <td>9</td>\n",
       "      <td>7</td>\n",
       "      <td>4</td>\n",
       "      <td>2</td>\n",
       "      <td>48</td>\n",
       "      <td>67</td>\n",
       "      <td>13534.787</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno      date  num_25  num_50  \\\n",
       "0  555bSTll4Rzaz1vBg/VfGfaXKEE8S74EAdKPOCkBqpg=  20170301       2       0   \n",
       "1  KORJwjTctoWENM9oM2Rrl432wxllmC1RvP5p84PUFI8=  20170316       6       1   \n",
       "2  +GdZIQJdsQSeRKyu/GONhgWzK4R8Ufm59RpzwAj4OCE=  20170310       5       1   \n",
       "3  9yZc5dve4sg96RQT1FyTPSmcDWrIVu+qsI7W7oOmFvE=  20170306       0       0   \n",
       "4  Rq9NMOw9RAbN3qUIG1MUU+vL0lqhAJUAsZiW0dsm/Lg=  20170322       9       7   \n",
       "\n",
       "   num_75  num_985  num_100  num_unq  total_secs  \n",
       "0       0        1       25       26    6306.242  \n",
       "1       2        3       36       34   10718.255  \n",
       "2       0        0       21       21    4948.548  \n",
       "3       1        0       28        1    3436.411  \n",
       "4       4        2       48       67   13534.787  "
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/logs.csv').head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The final dataset has about 400 million rows. We can read it in 10 million rows at a time. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "390000000.0 rows processed.\n",
      "Overall time: 3819 seconds.\n"
     ]
    }
   ],
   "source": [
    "chunksize = 1e7\n",
    "\n",
    "start = timer()\n",
    "\n",
    "for i, chunk in enumerate(pd.read_csv('/data/churn/user_logs.csv', chunksize = chunksize)):\n",
    "    partition_by_hashing(chunk, name = 'logs', progress = None)\n",
    "    \n",
    "    if (i + 1) % 10 == 0:\n",
    "        print(f'{i * chunksize} rows processed.', end = '\\r')\n",
    "    \n",
    "end = timer()\n",
    "print(f'\\nOverall time: {round(end - start)} seconds.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With the grouping approach, we were able to get 30 GB of data partitioned in less than 2 hours! Dealing with large datasets can be tough, but breaking down the problem makes it much more manageable. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>date</th>\n",
       "      <th>num_25</th>\n",
       "      <th>num_50</th>\n",
       "      <th>num_75</th>\n",
       "      <th>num_985</th>\n",
       "      <th>num_100</th>\n",
       "      <th>num_unq</th>\n",
       "      <th>total_secs</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>402013</th>\n",
       "      <td>fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=</td>\n",
       "      <td>20160509</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>15</td>\n",
       "      <td>15</td>\n",
       "      <td>4187.548</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>402014</th>\n",
       "      <td>fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=</td>\n",
       "      <td>20160703</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>4</td>\n",
       "      <td>4</td>\n",
       "      <td>924.083</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>402015</th>\n",
       "      <td>fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=</td>\n",
       "      <td>20161004</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>24</td>\n",
       "      <td>26</td>\n",
       "      <td>6031.879</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>402016</th>\n",
       "      <td>fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=</td>\n",
       "      <td>20170101</td>\n",
       "      <td>0</td>\n",
       "      <td>3</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>5</td>\n",
       "      <td>790.686</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>402017</th>\n",
       "      <td>xVJ8UUCOfXv5dXwxjFP5ffOHP+kgRRZbM7Mf5hJe0AE=</td>\n",
       "      <td>20150126</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>3</td>\n",
       "      <td>321.985</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                msno      date  num_25  \\\n",
       "402013  fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=  20160509       0   \n",
       "402014  fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=  20160703       1   \n",
       "402015  fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=  20161004       2   \n",
       "402016  fBNdslCoBSVgRLCTT/1wz2u5BJE0D4TuF+8g0+IPOLQ=  20170101       0   \n",
       "402017  xVJ8UUCOfXv5dXwxjFP5ffOHP+kgRRZbM7Mf5hJe0AE=  20150126       2   \n",
       "\n",
       "        num_50  num_75  num_985  num_100  num_unq  total_secs  \n",
       "402013       0       0        0       15       15    4187.548  \n",
       "402014       0       0        0        4        4     924.083  \n",
       "402015       0       0        0       24       26    6031.879  \n",
       "402016       3       2        0        1        5     790.686  \n",
       "402017       0       0        0        1        3     321.985  "
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.read_csv(base_dir + f'p{partition}/logs.csv').tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusions\n",
    "\n",
    "In this notebook, we implemented a partitioning of a dataset that would normally be too large to fit in memory. After trying several options, we eventually decided on the fastest process which was:\n",
    "\n",
    "1. Map customer ids to a partition using a hashing function.\n",
    "2. Compute the integer hash and then divide by the number of partitions \n",
    "3. Group the dataframe by the partition and write each partition to appropriate directory and file\n",
    "4. For large files that cannot fit all in memory, read in via chunking and send each chunk through the partitioning function\n",
    "\n",
    "Now we can work on an individual partition to develop an automated feature engineering pipeline. After the pipeline has been developed, we can use a framework such as Spark or Dask to run the partitions through the pipeline in parallel. This will speed up the overall feature engineering process and allow us to scale to larger datasets.\n",
    "\n",
    "## Next Steps\n",
    "\n",
    "To implement a machine learning solution, we need to take several steps outlined in the following process:\n",
    "\n",
    "1. Prediction Engineering: define a business need and translate into a machine learning problem. Create a set of labeled historical examples (called label times) that can be used to build features for each label.\n",
    "2. Use the label times to build features for each label by filtering the data to times before the cutoff time. This procedure can be rapidly completed using automated feature engineering.\n",
    "3. Train a machine learning algorithm to predict the labels from the features. Once the model has been optimized, use it to make predictions on new data.\n",
    "\n",
    "With the data in partitions, the first two steps can be done rapidly in parallel. The first step is implemented in the `Prediction Engineering` notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
